package com.syf.disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import net.openhft.affinity.AffinityThreadFactory;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.concurrent.ThreadFactory;

/**
 * 【7】1.创建Disruptor对象  -》 自动创建一个RingBuffer
 *      2.自动创建DisruptorTemplate 对象
 */
@Configuration
@EnableConfigurationProperties(value = DisruptorProperties.class)
public class DisruptorAutoConfiguration {

    // Disruptor配置属性类
    public DisruptorProperties disruptorProperties;
    public DisruptorAutoConfiguration(DisruptorProperties disruptorProperties) {
        this.disruptorProperties = disruptorProperties;
    }

    /**
     * 事件工厂：
     *      Disruptor 通过 EventFactory 在 RingBuffer 中预创建 Event
     *
     *      一个 Event 实例实际上被用作一个“数据槽”，发布者 在发布前，先从 RingBuffer 获得一个 Event 的实例，
     *      然后往 Event 实例中填充数据，之后再发布到 RingBuffer 中，
     *      之后由 Consumer 获得该 Event 实例并从中读取数据。
     */
    @Bean
    public EventFactory<OrderEvent> eventEventFactory() {

        EventFactory<OrderEvent> orderEventEventFactory = new EventFactory<OrderEvent>() {
            @Override
            public OrderEvent newInstance() {
                return new OrderEvent();  //从 RingBuffer 中获得一个 Event(数据槽)
            }
        };
        return orderEventEventFactory;
    }

    //线程工厂 -- 用于事件处理的线程池(提供的线程来触发 Consumer的消费事件处理)
    @Bean
    public ThreadFactory threadFactory() {
        return new AffinityThreadFactory("Match-Handler:") ;
    }

    /**
     * 无锁高效的等待策略：
     *      //1.是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现
     * 		WaitStrategy BLOCKING_WAIT = new BlockingWaitStrategy();
     * 		//2.性能表现跟BlockingWaitStrategy差不多，对CPU的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景
     * 		WaitStrategy SLEEPING_WAIT = new SleepingWaitStrategy();
     * 		//3.性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于CPU逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性
     * 		WaitStrategy YIELDING_WAIT = new YieldingWaitStrategy();
     */
    @Bean
    public WaitStrategy waitStrategy() {
        return new YieldingWaitStrategy();
    }

    /**
     * 1.自动创建一个RingBuffer 对象
     *      eventFactory: 事件工厂
     *      threadFactory: 我们执行者(消费者)的线程该怎么创建
     *      waitStrategy : 等待策略: 当我们ringBuffer 没有数据时,我们怎么等待
     *      eventHandlers： 事件处理器--多个EventHandler接口实现类(即多个消费者、多个线程处理)
     */
    @Bean
    public RingBuffer<OrderEvent> ringBuffer(
            EventFactory<OrderEvent> eventFactory,
            ThreadFactory threadFactory,
            WaitStrategy waitStrategy,
            EventHandler<OrderEvent>[] eventHandlers   //多个EventHandler的实现类 OrderEventHandler ，对应多个symbol交易对标识
    ) {

        //1）构建disruptor
        Disruptor<OrderEvent> disruptor = null;
        ProducerType producerType = ProducerType.SINGLE;   //生产类型：单生产者
        if (disruptorProperties.isMultiProducer()) {
            producerType = ProducerType.MULTI;  //支持多生产者
        }
        disruptor = new Disruptor<OrderEvent>(eventFactory, disruptorProperties.getRingBufferSize(), threadFactory, producerType, waitStrategy);
        disruptor.setDefaultExceptionHandler(new DisruptorHandlerException());  //设置Disruptor异常处理器

        //2）连接消费事件方法(设置消费者--每个消费者代表我们的一个交易对(Order)):
        //      有多少个交易对,我们就有多少个eventHandlers(线程) ,事件来了后,多个eventHandlers 是并发执行的
        disruptor.handleEventsWith(eventHandlers);

        //3）获取ringBuffer对象【高速缓存--OrderEvent】
        RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();

        //4）启动--开始监听
        disruptor.start();

        //5）使用优雅的停机
        final Disruptor<OrderEvent> disruptorShutdown = disruptor;
        Runtime.getRuntime().addShutdownHook(new Thread(
                () -> {
                    disruptorShutdown.shutdown();
                }, "DisruptorShutdownThread"  //ThreadName
        ));
        return ringBuffer;
    }


    /**
     * 2.自动创建DisruptorTemplate 对象
     *
     * @param ringBuffer
     * @return
     */
    @Bean
    public DisruptorTemplate disruptorTemplate(RingBuffer<OrderEvent> ringBuffer) {
        return new DisruptorTemplate(ringBuffer);
    }
}
